package com.mai.realtime.app.dwd.db;

import com.mai.realtime.app.BaseSqlApp;
import com.mai.realtime.common.Constant;
import com.mai.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Dwd_CourseTestPaper extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_CourseTestPaper().init(
                3002,
                3,
                "Dwd_CourseTestPaper"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 关联状态保存时间TTL
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(10));


        // 使用kafka连接器获取ods_db的动态表
        readOdsDb(tEnv,"Dwd_CourseTestPaper");


        // 从动态表读取测验问题表
        Table testPaperQuestion = tEnv.sqlQuery("select " +
                " data['id'] id , " +
                " data['paper_id'] paper_id ," +
                " data['question_id'] question_id , " +
                " data['score'] score , " +
                " data['create_time'] create_time ," +
                " data['deleted'] deleted ," +
                " data['publisher_id'] publisher_id ," +
                " ts" +
                " from ods_db" +
                " where `database`='gmall' " +
                " and `table`='test_paper_question' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("testPaperQuestion",testPaperQuestion);



        // 从动态表读取测验问题表
        Table testPaper = tEnv.sqlQuery("select " +
                " data['id'] id , " +
                " data['paper_title'] paper_title ," +
                " data['course_id'] course_id , " +
                " data['create_time'] create_time , " +
                " data['update_time'] update_time ," +
                " data['publisher_id'] publisher_id ," +
                " data['deleted'] deleted ," +
                " from ods_db" +
                " where `database`='edu' " +
                " and `table`='test_paper' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("testPaper",testPaper);


        // 从动态表读取测验问题表
        Table testQuestionOption = tEnv.sqlQuery("select " +
                " data['id'] id , " +
                " data['paper_title'] paper_title ," +
                " data['course_id'] course_id , " +
                " data['create_time'] create_time , " +
                " data['update_time'] update_time ," +
                " data['publisher_id'] publisher_id ," +
                " data['deleted'] deleted ," +
                " from ods_db" +
                " where `database`='edu' " +
                " and `table`='test_question_option' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("test_question_option",testQuestionOption);



        //TODO 从动态表读取测验问题表
        Table testExamQuestion = tEnv.sqlQuery("select " +
                " data['id'] id , " +
                " data['paper_id'] paper_id ," +
                " data['question_id'] question_id , " +
                " data['user_id'] user_id , " +
                " data['answer'] answer ," +
                " data['is_correct'] is_correct " +
                " from ods_db" +
                " where `database`='edu' " +
                " and `table`='test_exam_question' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("test_exam_question",testExamQuestion);



        //TODO 三个表关联
        Table result = tEnv.sqlQuery("select " +
                " t1.id," +
                " t1.paper_id," +
                " t1.question_id," +
                " t1.score," +
                " t2.paper_title," +
                " t2.course_id," +

                " t2.create_time," +
                " t2.update_time," +

                " t4.user_id," +

                " t4.answer," +
                " t4.is_correct," +
                " t1.ts " +
                " from testPaperQuestion t1" +
                " join testPaper t2" +
                " on t1.paper_id=t2.id" +
                " join test_exam_question t4 " +
                " on  t1.`paper_id`=t4.`paper_id` AND t1.`question_id`=t4.question_id  " +
                "");






        //TODO 定义动态表输出topic关联
        tEnv.executeSql("create table dwd_course_test_paper(" +
                " id string," +
                " paper_id string," +
                " question_id string," +
                " score string," +
                " paper_title string," +
                " course_id string," +
                " create_time string," +
                " update_time string," +
                " user_id string," +
                " answer string," +
                " is_correct string," +
                " ts BIGINT" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_COURSE_TEST_PAPER)
        );



        // 将结果输出DWD
        result.executeInsert("dwd_course_test_paper");
        //result.execute().print();



    }
}
